<?php
namespace app\service;

use think\Db;
use think\facade\Log;
use think\facade\Cache;
use app\service\MessageParser\MessageParserFactory;

/**
 * MQTT消息处理类
 * 专门用于处理接收到的MQTT消息并保存到flow_wifi_imei表
 */
class MqttMessageHandler
{
    /**
     * 消息批处理大小
     * 当消息达到此数量时批量插入数据库
     */
    protected $batchSize = 100;

    /**
     * 消息缓冲区
     */
    protected $messageBuffer = [];

    /**
     * 上次保存时间
     */
    protected $lastSaveTime = 0;

    /**
     * 最大缓冲时间(秒)
     * 即使未达到批处理大小，超过此时间也会保存
     */
    protected $maxBufferTime = 5;

    /**
     * 使用的数据库连接名称
     * 默认使用主数据库，可以设置为 'db_config2' 使用第二个数据库
     */
    protected $dbConnection = null; // null 表示使用默认数据库

    /**
     * IMEI缓存过期时间（秒）
     * 30分钟 = 1800秒
     */
    protected $cacheExpire = 1800;

    /**
     * 缓存键前缀
     */
    protected $cachePrefix = 'imei_exists:';

    /**
     * 数据库连接实例缓存
     */
    protected $dbInstance = null;

    /**
     * 连接重试次数
     */
    protected $maxRetries = 3;

    /**
     * 当前重试次数
     */
    protected $currentRetries = 0;

    /**
     * 是否启用析构日志
     * 设置为false可以避免在生产环境中产生过多日志
     */
    protected $enableDestructorLog = false;

    /**
     * 设置数据库连接
     * @param string|null $connection 数据库连接名称，null表示使用默认连接
     */
    public function setDbConnection($connection = null)
    {
        $this->dbConnection = $connection;
        $this->dbInstance = null; // 重置连接实例
        
    }

    /**
     * 获取数据库连接实例（带连接池优化）
     * @return \think\db\Query
     */
    protected function getDbInstance()
    {
        if ($this->dbInstance === null) {
            try {
                if ($this->dbConnection) {
                    $this->dbInstance = Db::connect($this->dbConnection);
                } else {
                    $this->dbInstance = Db::class;
                }
                $this->currentRetries = 0; // 重置重试次数
            } catch (\Throwable $e) {
                
                throw $e;
            }
        }
        return $this->dbInstance;
    }

    /**
     * 执行数据库操作（带重连机制）
     * @param callable $operation 数据库操作回调
     * @param string $operationName 操作名称（用于日志）
     * @return mixed 操作结果
     */
    protected function executeDbOperation(callable $operation, string $operationName = 'database operation')
    {
        $lastException = null;
        for ($i = 0; $i <= $this->maxRetries; $i++) {
            try {
                $result = $operation();
                
                return $result;
            } catch (\Throwable $e) {
                $lastException = $e;
                $this->currentRetries = $i + 1;
                

                // 检查是否是连接相关的错误
                if ($this->isConnectionError($e) && $i < $this->maxRetries) {
                    
                    $this->dbInstance = null; // 重置连接实例
                    usleep(100000 * ($i + 1)); // 递增延迟：100ms, 200ms, 300ms
                    continue;
                } else {
                    // 非连接错误或重试次数用完，直接抛出异常
                    
                    throw $e;
                }
            }
        }

        throw $lastException;
    }

    /**
     * 判断是否是连接相关的错误
     * @param \Throwable $e 异常
     * @return bool 是否是连接错误
     */
    protected function isConnectionError(\Throwable $e): bool
    {
        $message = strtolower($e->getMessage());
        $connectionErrors = [
            'server has gone away',
            'lost connection',
            'connection refused',
            'connection reset',
            'connection timeout',
            'connection failed',
            'broken pipe',
            'no such file or directory'
        ];

        foreach ($connectionErrors as $error) {
            if (strpos($message, $error) !== false) {
                return true;
            }
        }

        return false;
    }

    /**
     * 检查IMEI是否存在（使用缓存优化）
     * @param string $imei IMEI号码
     * @return bool 是否存在
     */
    public function checkImeiExists($imei): bool
    {
        $cacheKey = $this->cachePrefix . $imei;

        // 先查缓存
        $cached = Cache::get($cacheKey);
        if ($cached !== null) {
            
            return (bool)$cached;
        }

        // 缓存未命中，查询数据库
        try {
            if ($this->dbConnection) {
                $existing = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                    ->where('IMEI', $imei)
                    ->find();
            } else {
                $existing = \think\Db::name('flow_wifi_imei')
                    ->where('IMEI', $imei)
                    ->find();
            }

            $exists = !empty($existing);

            // 将结果存入缓存
            Cache::set($cacheKey, $exists ? 1 : 0, $this->cacheExpire);
            

            return $exists;
        } catch (\Throwable $e) {
            
            return false;
        }
    }

    /**
     * 批量检查IMEI是否存在（缓存优化）
     * @param array $imeis IMEI数组
     * @return array 存在的IMEI列表
     */
    protected function batchCheckImeiExists($imeis): array
    {
        // 检查是否为CLI环境，如果Redis有问题则直接查询数据库
        if (php_sapi_name() === 'cli') {
            try {
                // 先测试一下缓存是否可用
                Cache::set('cli_test_' . time(), 'test', 1);
            } catch (\Throwable $e) {
                
                return $this->directCheckImeiExists($imeis);
            }
        }

        $existingImeis = [];
        $uncachedImeis = [];

        // 先批量查询缓存
        foreach ($imeis as $imei) {
            $cacheKey = $this->cachePrefix . $imei;

            $cached = false;
            try {
                $cached = Cache::get($cacheKey);
            } catch (\Throwable $e) {
                // 缓存有问题，直接查询数据库
                
                $uncachedImeis[] = $imei;
                continue;
            }

            

            if ($cached !== false) {  // 缓存存在（不管是0还是1）
                if ($cached == 1) {   // 明确检查是否为1（存在）
                    $existingImeis[] = $imei;
                } else {
                    // 缓存显示不存在，但为了确保准确性，重新查询数据库
                    $uncachedImeis[] = $imei;
                }
            } else {  // 缓存不存在（返回false）
                $uncachedImeis[] = $imei;
            }
        }

        $hitCount = count($imeis) - count($uncachedImeis);
        $totalCount = count($imeis);
        // 批量查询未缓存的IMEI
        if (!empty($uncachedImeis)) {

            try {
                
                if ($this->dbConnection) {
                    $dbResults = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                        ->where('IMEI', 'in', $uncachedImeis)
                        ->column('IMEI');
                } else {

                    $dbResults = \think\Db::name('flow_wifi_imei')
                        ->where('IMEI', 'in', $uncachedImeis)
                        ->column('IMEI');
                }

                

                // 更新缓存并合并结果
                foreach ($uncachedImeis as $imei) {
                    $exists = in_array($imei, $dbResults);
                    $cacheKey = $this->cachePrefix . $imei;

                    // 尝试更新缓存，但不影响主要逻辑
                    try {
                        Cache::set($cacheKey, $exists ? 1 : 0, $this->cacheExpire);
                        
                    } catch (\Throwable $e) {
                        try {
                            cache($cacheKey, $exists ? 1 : 0, $this->cacheExpire);
                            
                        } catch (\Throwable $e2) {
                            
                        }
                    }

                    // 无论缓存是否成功，都要处理数据库查询结果
                    if ($exists) {
                        $existingImeis[] = $imei;
                    } else {
                    }
                }

                $uncachedCount = count($uncachedImeis);
                $foundCount = count($dbResults);
            } catch (\Throwable $e) {
                
                // 即使数据库查询失败，也要尝试处理已有的结果
            }
        }

        return $existingImeis;
    }

    /**
     * 直接查询数据库检查IMEI是否存在（不使用缓存）
     * @param array $imeis IMEI数组
     * @return array 存在的IMEI列表
     */
    protected function directCheckImeiExists(array $imeis): array
    {
        try {
            

            if ($this->dbConnection) {
                $existingImeis = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                    ->where('IMEI', 'in', $imeis)
                    ->column('IMEI');
            } else {
                $existingImeis = \think\Db::name('flow_wifi_imei')
                    ->where('IMEI', 'in', $imeis)
                    ->column('IMEI');
            }

            

            return $existingImeis;

        } catch (\Throwable $e) {
            
            return [];
        }
    }

    /**
     * 清除指定IMEI的缓存
     * @param string $imei IMEI号码
     */
    public function clearImeiCache($imei)
    {
        $cacheKey = $this->cachePrefix . $imei;
        try {
            Cache::rm($cacheKey);
        } catch (\Throwable $e) {
            cache($cacheKey, null);
        }
    }

    /**
     * 清除所有IMEI缓存
     */
    public function clearAllImeiCache()
    {
        // 这里可以根据实际需要实现
        // 由于Redis没有直接的通配符删除，可以考虑使用标签或其他方式
    }

    /**
     * 强制重新检查IMEI是否存在（忽略缓存）
     * @param string $imei IMEI号码
     * @return bool 是否存在
     */
    public function forceCheckImeiExists($imei): bool
    {
        try {
            // 先清除缓存
            $this->clearImeiCache($imei);

            // 直接查询数据库
            if ($this->dbConnection) {
                $exists = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                    ->where('IMEI', $imei)
                    ->count() > 0;
            } else {
                $exists = \think\Db::name('flow_wifi_imei')
                    ->where('IMEI', $imei)
                    ->count() > 0;
            }

            // 更新缓存
            $cacheKey = $this->cachePrefix . $imei;
            try {
                Cache::set($cacheKey, $exists ? 1 : 0, $this->cacheExpire);
            } catch (\Throwable $e) {
                cache($cacheKey, $exists ? 1 : 0, $this->cacheExpire);
            }

            

            return $exists;

        } catch (\Throwable $e) {
            
            return false;
        }
    }

    /**
     * 批量更新记录
     * @param array $updateData 要更新的数据数组
     * @return int 成功更新的记录数
     */
    protected function batchUpdateRecords(array $updateData): int
    {

        if (empty($updateData)) {
            return 0;
        }

        $successCount = 0;

        // 按IMEI分组，避免重复更新
        $groupedData = [];
        foreach ($updateData as $data) {
            $imei = $data['IMEI'];
            $groupedData[$imei] = $data; // 后面的数据会覆盖前面的
        }

        // 构建批量更新SQL
        try {
            $db = $this->dbConnection ? Db::connect($this->dbConnection) : Db::name('flow_wifi_imei');

            // 对于MySQL，使用CASE WHEN语句进行批量更新
            if (count($groupedData) > 1) {
                $successCount = $this->buildBatchUpdateSql($groupedData);
            } else {

                // 单条记录直接更新
                $data = reset($groupedData);

                $imei = $data['IMEI'];
                unset($data['IMEI']); // 移除IMEI，避免更新主键

                if ($this->dbConnection) {

                    $result = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                        ->where('IMEI', $imei)
                        ->update($data);
                } else {

                    $result = Db::name('flow_wifi_imei')
                        ->where('IMEI', $imei)
                        ->update($data);

                }

                $successCount = $result !== false ? 1 : 0;
            }

        } catch (\Throwable $e) {
            
            throw $e;
        }

        return $successCount;
    }

    /**
     * 构建批量更新SQL
     * @param array $groupedData 按IMEI分组的数据
     * @return int 成功更新的记录数
     */
    protected function buildBatchUpdateSql(array $groupedData): int
    {
        if (empty($groupedData)) {
            return 0;
        }

        // 获取所有需要更新的字段
        $allFields = [];
        foreach ($groupedData as $data) {
            $allFields = array_merge($allFields, array_keys($data));
        }
        $allFields = array_unique($allFields);
        $allFields = array_diff($allFields, ['IMEI']); // 排除IMEI字段

        if (empty($allFields)) {
            return 0;
        }

        // 构建CASE WHEN语句
        $caseClauses = [];
        $imeis = array_keys($groupedData);

        foreach ($allFields as $field) {
            $caseClause = "`{$field}` = CASE `IMEI`";
            foreach ($groupedData as $imei => $data) {
                if (isset($data[$field])) {
                    $value = is_string($data[$field]) ? "'" . addslashes($data[$field]) . "'" : $data[$field];
                    $caseClause .= " WHEN '{$imei}' THEN {$value}";
                }
            }
            $caseClause .= " ELSE `{$field}` END";
            $caseClauses[] = $caseClause;
        }

        // 构建完整的UPDATE SQL
        $sql = "UPDATE `flow_wifi_imei` SET " . implode(', ', $caseClauses) .
               " WHERE `IMEI` IN ('" . implode("','", $imeis) . "')";

        return $this->executeDbOperation(function() use ($sql) {
            if ($this->dbConnection) {
                $result = Db::connect($this->dbConnection)->execute($sql);
            } else {
                $result = Db::execute($sql);
            }
            return $result;
        }, "批量更新SQL");
    }

    /**
     * 降级为逐条更新
     * @param array $updateData 要更新的数据数组
     * @return int 成功更新的记录数
     */
    protected function fallbackSingleUpdate(array $updateData): int
    {
        $successCount = 0;

        foreach ($updateData as $data) {
            try {
                $imei = $data['IMEI'];
                unset($data['IMEI']); // 移除IMEI，避免更新主键

                if ($this->dbConnection) {
                    $result = Db::connect($this->dbConnection)->name('flow_wifi_imei')
                        ->where('IMEI', $imei)
                        ->update($data);
                } else {
                    $result = Db::name('flow_wifi_imei')
                        ->where('IMEI', $imei)
                        ->update($data);
                }

                if ($result !== false) {
                    $successCount++;
                }
            } catch (\Throwable $e) {
                
            }
        }
        return $successCount;
    }

    /**
     * 处理消息
     * @param string $topic 主题
     * @param array $messageData 消息数据
     * @return bool 处理结果
     */
    public function handle($topic, $messageData): bool
    {
        try {
            // 确保消息数据是数组格式
            if (is_string($messageData)) {
                $messageData = json_decode($messageData, true);
                if (!is_array($messageData)) {
                    return false;
                }
            }

            // 使用MessageParserFactory自动检测协议
            $parser = MessageParserFactory::detect($topic, $messageData);

            if ($parser) {
                // 使用检测到的解析器解析消息
                $parsedData = $parser->parse($topic, $messageData);

                if ($parsedData) {
                    // 将原始消息数据附加到解析数据中，供后续日志记录使用
                    $parsedData['_original_topic'] = $topic;
                    $parsedData['_original_message'] = $messageData;

                    // 添加到缓冲区
                    $this->addToBuffer($parsedData);
                    return true;
                }
            }

            return false;

        } catch (\Throwable $e) {
            return false;
        }
    }

    /**
     * 解析设备消息数据
     * @param array $data 原始消息数据
     * @param string $deviceMac 设备MAC地址
     * @return array|null 解析后的数据
     */
    protected function parseDeviceMessage($data, $deviceMac): ?array
    {
        try {
            // 判断协议类型：是否存在 hotspot_info_5g 字段
            $is5GProtocol = isset($data['hotspot_info_5g']);

            

            // 获取设备信息
            $deviceInfo = $data['device_info'] ?? [];
            $systemInfo = $data['system_info'] ?? [];
            $networkInfo = $data['network_info'] ?? [];
            $simInfo = $data['sim_info'] ?? [];

            // 根据协议类型获取热点信息

            if ($is5GProtocol) {
                // 5G协议：使用 hotspot_info 作为主要热点信息，hotspot_info_5g 作为5G热点信息
                $hotspotInfo = $data['hotspot_info_5g'] ?? [];
                $hotspotInfo5g = $data['hotspot_info_5g'] ?? [];
            } else {
                // 4G协议：只有 hotspot_info，没有 hotspot_info_5g
                $hotspotInfo = $data['hotspot_info'] ?? [];
                $hotspotInfo5g = []; // 4G协议没有5G热点信息
            }

            // 获取当前生效的SIM卡信息
            $currentSimId = $data['cur_sim_id'] ?? 1;
            $currentSim = $this->getCurrentSimInfo($simInfo, $currentSimId, $is5GProtocol);
            //wo zuoceshi
            //nihao
            // 准备保存的数据
            $saveData = [
                'IMEI' => $deviceInfo['dev_imei2'] ?? '',
                'mac' => $deviceMac, // 从主题中提取的MAC地址
                'iccid' => $currentSim['sim_iccid'] ?? '',
                'signal_strength' => $currentSim['sim_signal_level'] ?? 4,
                'mnc' => $this->getMncType($currentSim['sim_mnc'] ?? '00'),
                'ssid' => $hotspotInfo['hotspot_name'] ?? '',
                'key' => $hotspotInfo['hotspot_password'] ?? '',
//                'user_ssid' => $hotspotInfo5g['hotspot_name'] ?? '',
//                'user_key' => $hotspotInfo5g['hotspot_password'] ?? '',
                'main_sim' => $currentSimId,
                'user_sim' => $currentSimId,
                'band' => $this->getBandType($hotspotInfo, $hotspotInfo5g, $is5GProtocol),
                'hidden_ssid' => $hotspotInfo['hotspot_whether_hide'] ?? 0,
                'conn_cnt' => $hotspotInfo['hotspot_connected_num'] ?? 0,
                'remain_pwr' => $this->getBatteryLevel($systemInfo),
                'dbm' => $currentSim['sim_dbm'] ?? '',
                //'add_time' => date('Y-m-d H:i:s'),
                'update_time' => date('Y-m-d H:i:s'),
                //'api_update_time' => date('Y-m-d H:i:s'),
            ];

            // 根据SIM卡信息设置对应的ICCID
//            foreach ($simInfo as $sim) {
//                if ($sim['sim_exist'] == 1) {
//                    $simId = $sim['sim_id'];
//                    $iccid = $sim['sim_iccid'] ?? '';
//                    if ($simId >= 1 && $simId <= 4) {
//                        $saveData['wifi_iccid' . $simId] = $iccid;
//                    }
//                }
//            }

            // 根据运营商设置对应的号段字段
            if ($currentSim) {
                $operator = $this->getOperatorByMnc($currentSim['sim_mnc'] ?? '00');
                $phoneNumber = $this->extractPhoneNumber($currentSim['sim_iccid'] ?? '');

                switch ($operator) {
                    case 'mobile':
                        $saveData['yidong_kahao'] = $phoneNumber;
                        break;
                    case 'unicom':
                        $saveData['liangtong_kahao'] = $phoneNumber;
                        break;
                    case 'telecom':
                        $saveData['dianxin_kahao'] = $phoneNumber;
                        break;
                }
            }

            
            return $saveData;

        } catch (\Throwable $e) {
            return null;
        }
    }

    /**
     * 记录原始消息到日志表
     * @param string $topic MQTT主题
     * @param array $messageData 原始消息数据
     * @param string $imei 设备IMEI
     * @param string $iccid 设备ICCID
     */
    protected function logOriginalMessage($topic, $messageData, $imei, $iccid = ''): void
    {
        try {
            // 方法1：使用模型（推荐）
            $success = \app\model\MqttLog::logMessage($imei, $messageData, $iccid);

            if (!$success) {
                // 如果模型方法失败，尝试直接数据库操作
                try {
                    $logData = [
                        'imei' => $imei,
                        'iccid' => $iccid,
                        'content' => json_encode($messageData, JSON_UNESCAPED_UNICODE),
                        'createtime' => date('Y-m-d H:i:s')
                    ];

                    Db::connect('db_config2')->name('log')->insert($logData);
                } catch (\Throwable $e2) {
                    // 忽略日志记录失败
                }
            }
        } catch (\Throwable $e) {
            // 忽略异常
        }
    }

    /**
     * 获取当前SIM卡信息
     * @param array $simInfo SIM卡信息数组
     * @param int $currentSimId 当前SIM卡ID
     * @param bool $is5GProtocol 是否为5G协议
     * @return array|null 当前SIM卡信息
     */
    protected function getCurrentSimInfo($simInfo, $currentSimId, $is5GProtocol): ?array
    {
        foreach ($simInfo as $sim) {
            if ($sim['sim_id'] == $currentSimId && $sim['sim_exist'] == 1) {
                // 处理不同协议中字段的差异
                if (!$is5GProtocol) {
                    // 4G协议中，某些字段可能有不同的格式
                    // 例如：sim_dbm 在4G中可能是 "-71dBm" 格式，需要提取数值
                    if (isset($sim['sim_dbm']) && is_string($sim['sim_dbm'])) {
                        // 提取数值部分，去掉 "dBm" 后缀
                        $sim['sim_dbm'] = (int)filter_var($sim['sim_dbm'], FILTER_SANITIZE_NUMBER_INT);
                    }
                }
                return $sim;
            }
        }
        return null;
    }

    /**
     * 获取频段类型
     * @param array $hotspotInfo 热点信息
     * @param array $hotspotInfo5g 5G热点信息
     * @param bool $is5GProtocol 是否为5G协议
     * @return string 频段类型 ('a' 或 'g')
     */
    protected function getBandType($hotspotInfo, $hotspotInfo5g, $is5GProtocol): string
    {
        if ($is5GProtocol) {
            // 5G协议：检查 hotspot_info_5g 的 hotspot_whether_5g 字段
            return ($hotspotInfo5g['hotspot_whether_5g'] ?? 0) == 1 ? 'a' : 'g';
        } else {
            // 4G协议：检查 hotspot_info 的 hotspot_band 字段
            // hotspot_band: 0=2.4G, 1=5G
            return ($hotspotInfo['hotspot_band'] ?? 0) == 1 ? 'a' : 'g';
        }
    }

    /**
     * 获取电池电量
     * @param array $systemInfo 系统信息
     * @return int 电池电量
     */
    protected function getBatteryLevel($systemInfo): int
    {
        // 5G协议使用 sys_bat_level，4G协议也使用 sys_bat_level
        // 但4G协议中可能有拼写错误的字段名
        return $systemInfo['sys_bat_level'] ?? $systemInfo['sys_battery_level'] ?? 0;
    }

    /**
     * 根据MNC获取运营商类型
     * @param string $mnc MNC代码
     * @return string 运营商类型
     */
    protected function getOperatorByMnc($mnc): string
    {
        switch ($mnc) {
            case '00':
            case '02':
            case '07':
            case '08':
                return 'mobile'; // 中国移动
            case '01':
            case '06':
            case '09':
                return 'unicom'; // 中国联通
            case '03':
            case '05':
            case '11':
                return 'telecom'; // 中国电信
            default:
                return 'mobile'; // 默认移动
        }
    }

    /**
     * 获取MNC类型数字
     * @param string $mnc MNC代码
     * @return int 类型数字
     */
    protected function getMncType($mnc): int
    {
        switch ($mnc) {
            case '00':
            case '02':
            case '07':
            case '08':
                return 0; // 移动
            case '01':
            case '06':
            case '09':
                return 1; // 联通
            case '03':
            case '05':
            case '11':
                return 2; // 电信
            default:
                return 0; // 默认移动
        }
    }

    /**
     * 从ICCID中提取手机号码（简化处理）
     * @param string $iccid ICCID
     * @return string 手机号码
     */
    protected function extractPhoneNumber($iccid): string
    {
        // 这里是简化处理，实际可能需要更复杂的逻辑
        // ICCID通常不直接包含手机号码，这里返回ICCID的后11位作为示例
        if (strlen($iccid) >= 11) {
            return substr($iccid, -11);
        }
        return $iccid;
    }

    /**
     * 添加消息到缓冲区
     * @param array $data 消息数据
     */
    protected function addToBuffer(array $data): void
    {

        // 添加到缓冲区
        $this->messageBuffer[] = $data;

        

        // 检查是否需要保存
        $currentTime = time();

        $bufferFull = count($this->messageBuffer) >= $this->batchSize;
        $timeExceeded = ($currentTime - $this->lastSaveTime) >= $this->maxBufferTime && !empty($this->messageBuffer);

        if ($bufferFull) {
            $this->saveToDatabase();
            $this->lastSaveTime = $currentTime;
        } elseif ($timeExceeded) {
            $this->saveToDatabase();
            $this->lastSaveTime = $currentTime;
        } else {
        }
    }

    /**
     * 保存消息到数据库
     */
    protected function saveToDatabase(): void
    {

        if (empty($this->messageBuffer)) {
            return;
        }

        try {
            // 数据库连接优化：移除不必要的连接测试，直接执行业务SQL

            // 使用缓存优化的批量处理
            $successCount = 0;
            $skippedCount = 0;

            // 提取所有IMEI进行批量缓存查询
            $allImeis = array_column($this->messageBuffer, 'IMEI');

            $existingImeis = $this->batchCheckImeiExists($allImeis);
            // 只处理存在的 IMEI
            $updateData = [];

            foreach ($this->messageBuffer as $data) {
                $imei = $data['IMEI'];

                // 只处理存在的 IMEI，不存在的直接跳过
                if (!in_array($imei, $existingImeis)) {
                    continue;
                }

                // 记录原始消息到日志表
                if (isset($data['_original_topic']) && isset($data['_original_message'])) {
                    $iccid = $data['iccid'] ?? '';
                    $this->logOriginalMessage($data['_original_topic'], $data['_original_message'], $imei, $iccid);
                }

                // 移除临时字段
                unset($data['_original_topic'], $data['_original_message']);

                // 过滤空值并添加更新时间
                $cleanData = array_filter($data, function($value) {
                    return $value !== '' && $value !== null;
                });
                $cleanData['update_time'] = date('Y-m-d H:i:s');
                $cleanData['api_update_time'] = date('Y-m-d H:i:s');

                $updateData[] = $cleanData;
            }

            // 批量更新存在的记录
            if (!empty($updateData)) {
                try {
                    $successCount = $this->batchUpdateRecords($updateData);
                } catch (\Throwable $e) {
                    // 降级为逐条更新
                    $successCount = $this->fallbackSingleUpdate($updateData);
                }
            }

            // 清空缓冲区
            $this->messageBuffer = [];

        } catch (\Throwable $e) {
            // 失败后，保留一部分消息重试
            // 避免缓冲区无限增长
            if (count($this->messageBuffer) > $this->batchSize * 2) {
                $this->messageBuffer = array_slice($this->messageBuffer, -$this->batchSize);
            }
        }
    }

    /**
     * 强制保存所有缓冲的消息
     */
    public function flush(): void
    {
        $this->saveToDatabase();
    }

    /**
     * 设置是否启用析构日志
     * @param bool $enable 是否启用
     */
    public function setDestructorLog(bool $enable): void
    {
        $this->enableDestructorLog = $enable;
    }

    /**
     * 析构函数，确保所有消息都被保存
     */
    public function __destruct()
    {
        $this->flush();
    }
}
